object Sparkcore {
  /**
   * spark是基于内存的计算框架，在遇到内存不足时，可用mapreduce补充
   * Hadoop主要是处理一次性数据处理计算，使用的是文件系统存储（磁盘）
   * sparkCore提供了spark最核心基础的功能，其他sql,Stream,Graphx,mllib在此基础上扩展
   * spark sql:用来操作结构化数据的组件，用户可以基于他使用sql语言或者hive操作数据
   * spark streaming:对数据进行实时操作流式计算的组件
   * spark MLlib:机器学习算法库
   * GraphX:图形框架
   * */
  /**
   * spark的数据结构： RDD（弹性分布式数据集）,累加器（分布式共享只写变量），广播变量（分布式共享只读变量）
   * rdd看成最小计算单元
   *  */
  /**
   * makeRDD()：从集合中（内存中）创建RDD
   * RDD的并行度和分区:有几个核就默认分几个区
   * rdd.mapPartitions:可以以分区为单位进行数据转换操作，但是
   *                  会将整个分区的数据加载到内存中进行引用，处理完的数据不会被释放，
   *                  会存在溢出现象
   *rdd.map({}):里面一般是匿名函数或函数，将数据进行逐条映射转换，可以是类型或值转换
   *           一般用于数据结构转换：
   *            dataRDD.map(
                      line => {
                        val datas: Array[String] = line.split(" ")
                        ((datas(1), datas(4)), 1)
                      }
                    )
   * rdd.mapPartitions():将待处理的数据以分区为单位发送到计算节点进行处理（处理是任意处理）
   * 内存不够用时一般使用map操作
   *
   * rdd.mapPartitionWithIndex():将待处理的数据以分区为单位发送到计算节点进行处理，在处理的同时可以获取当前分区索引
   *rdd.flatMap():把数据或集合进行扁平化操作，即把里面的东西弄成单个单个的
   *            如果有多个混合类型的数据，则进行match匹配
   *rdd.glom():将同一个分区的数据直接转换为相同类型的内存数据进行处理，分区不变
   *           函数里面不用进行任何转换，直接调用就行,之后可以进行数据大小比较，也可以进行分区求和
   *rdd.groupBy():将数据根据指定规则进行分组，分区默认不变，但会将数据打乱并重新组合（称为shuffle）,
   *            一个组的数据在一个分区中，但并不是说一个分区中只有一个组，分组函数会根据计算出来的key进行分组
   *rdd.filter()：根据指定规则对数据进行过滤，符合规则的数据保留，数据帅选过滤后，分区不变，但是分区内的数据可能会变得不均衡
   *         产生数据倾斜
   /*rdd.sample():根据指定的规则从数据集中抽取部分数据
   *     sample算子需要传递三个参数
   *    第一个参数表示抽取数据后是否将数据返回true(放回)或者false
   *    第二参数表示每条数据会被抽取的概率,如果抽取不放回，每条数据中被抽取的概率基准值相同，如果抽取放回，概率表示每条数据被抽取的可能次数
   *    第三个参数表示被抽取数据随机算法的种子，如果不传递第三个参数，那么使用的是当前系统时间
   *
   *rdd.distinct():将数据集中重复的数据去重，只要是相同的就可去除，包括null
   *
   *rdd.coalesce():用于缩减分区，用于大数据集过滤后，提高小数据集的执行效率（存在过多的小任务时，可以通过此方法，
   *         收缩合并分区，减少分区的个数，减少调度成本）。coalesce()方法默认不会将数据打乱进行重新组合的，如果想要数据均衡，可以调用groupBy()
   *         对数据进行shuffle重新组合操作，也可以在参数中加入true。可以扩大分区扩大分区的同时需要进行shuffle操作才能起作用(即在参数中加入true)
   *repartition():扩大分区，底层采用的是coalesce(),并同时进行shuffle操作
   *缩减分区：用coalesce(),想要数据均衡，可以加上shuffle操作
   * 扩大分区：用repartition()操作，底层调用的是coalesce()操作，并shuffle操作
   *
   * sortBy(_._1)：排序数据，在排序数据之前，可通过函数方法进行处理，然后对结果进行排序，默认是升序排序
   *          排序产生的RDD分区数和原RDD的分区数一致，中间存在shuffle过程
   *          sortBy使用是通常需要将数据转换成List集合或其他形式的结构，iter.toList
   *          sortBy(_._2)表示排序按照元素中的第二个数据来排，默认是升序，如果要降序，则加上(Ordering.Int.reverse)
   *  take(3)表示取值取前3个值
   *
   *两个数据源中数据集操作：
   *     交集：intersection():对源RDD中的数据源进行求交集操作
   *           val rddjaio: RDD[Int] = rdd1.intersection(rdd2)
               println(rddjaio.collect().mkString(","))
   *     合并：union():对两个源RDD中的数据进行合并操作，得到一个新的RDD集合
   *     差集：subtract():对两个源RDD中的数据集进行差集操作，即去除重复的数据集，保留其他元素
   *     拉链：zip():将两个数据源中的数据以key-value的形式进行合并，第一个集合中的值作为key,第二个集合中的值作为value
   *交集，并集和差集要求两个数据源中的数据必须相同
   * 拉链中的数据源可以不相同 ，分区数，数据数量要求一致
   *
   *partitionBy():根据指定的分区规则对数据进行重新分区，Spark的默认分区器是HashPartitioner
   *      val rdd2:RDD[(Int,String)]=rdd.partitionBy(new HashPartitioner(2))
   *rdd.reduceByKey():把相同的key数据进行value数据聚合操作，即key相同，value值进相关的操作
   *         reduceByKey中如果key的数据只有一个，是不会参与聚合操作的，会直接返回
   *rdd.groupByKey():相同的key会被分配到一个组中，将数据源中的数据，相同key的数据分在一个组中，形成一个对偶元组
   *             元组中的第一个元素就是key，元组中的第二个元素就是相同key的value的集合
   *groupByKey会使数据打乱重新组合，存在shuffle操作，只能分组，轮盘之后再分组
   *reduceByKey：先对分区内的数据进行key值相同的数据进行聚合，然后再进行shuffle操作，性能比groupByKey好，分区内和分区间的规则要相同
   *spark中数据进行重新组合时，即shuffle操作时，必须进行落盘处理（数据写在磁盘上），不能在内存中等待，否则会导致内存溢出
   * mapValues():将key-value中的数据取出来，只取values，不取key
   * //分区内和分区间的计算常用算子：
   *aggregateByKey():将数据根据不同的规则进行分区内计算和分区间计算，存在两个柯里化的参数列表：
   *     第一个参数列表表示：
   *          需要传递一个参数，表述为初始值，主要用于当碰见第一个key时，和value进行分区内计算
   *     第二个参数列表表示：
   *          第一个参数表示分区内的计算操作
   *          第二个参数表示分区间的计算操作
   *aggregateByKey()中的初始值只会参与分区内计算
   *aggregate():初始值会参与分区内计算，并且分区内计算结果会参与分区间计算
   *
   *flodByKey()(,):如果聚合计算时，分区内和分区间计算规则相同，spark提供了flodByKey()(,)用于简化，参数只有初始值和计算规则的值
   *combineByKey{}:通常是把参与运算的第一个数做一些格式上的转换，需要三个参数，并且转换过程需要进行动态识别，所以要加上参数类型
   *       第一个参数表示：将相同key的第一个数据进行结构的转换，实现操作
   *       第二个参数：分区内的计算规则
   *       第三个参数表示：分区间的计算规则
   *reduceByKey()，agregateByKey(),flodByKey()(,),combineByKey{}的区别：
   *      都可以用来做聚合函数功能的操作。底层调用都一样、
   *      底层都调用combineByKeyWithClassTag(
   *      createCombiner,//相同key的第一条数据进行的处理函数
   *      mergeValue,//表示分区内数据的函数处理
   *      mergeCombiners//表示分区间数据的处理函数
   *      )
   *
   * //链接：没有匹配到的就不会显示出来
   * join():将具有相同类型的key的数据的值进行拼接，
   *       返回的是一个相同key对应的所有元素来凝结在一起的(k,(k,v))的RDD
   *       两个数据源中的key没有匹配上，那么数据不会出现在结果中
   *       两个数据源中有多个相同的key，则会依次匹配，可能会出现笛卡尔集，数量会几何性增长
   *leftOuterJoin():类似于sql语句的左外链接（左边有数据，右边可有可无）,没有匹配的则为空
   *rightOuterJoin():右外链接
   *
   *cogroup():connect+group，相同的key放在同一个组中然后链接在一起,可以同时操作三个rdd，
   *         多个数据链接一般用cogroup()，其既有分组的概念，又有链接的概念。
   *
   *
   */
    *
 //行动算子：能自动触发作业（Job）执行的方法
   /*行动算子:
        底层调用的都是runJob方法，底层会创建ActiveJob()
   collect():将不同分区的数据按照分区数据采集到Driver中，形成数组
   reduce():直接执行相关数据的操作
   count():rdd数据源中数据的个数
   first():获取数据源中的第一个数据
   take():获取N个数据
   takeOrdered():数据排序后，取N个数据
   aggregate(0)(,):和aggregateByKey()算子用法类似，参数需要三个，但是直接触发的
      *aggregateByKey()中的初始值只会参与分区内计算
      *aggregate():初始值会参与分区内计算，并且分区内计算结果会参与分区间计算
   fold():aggregate的简化操作
   countByValue():统计列表中某个值出现的次数，返回Map形式
   countByKey():统计key-value列表中key的出现次数，返回Map形式
   collect().foreach():是Driver端你存集合的循环遍历方法
   foreach():单独的是Executor端内存数据打印,分布式遍历RDD中的每一个元素，调用指定函数

    **/
    *
    /**
   实现数据统计的方法统计：即类似于WordCount操作
      1、groupBy():将数据根据指定规则进行分组，分区默认不变，但会将数据打乱并重新组合（称为shuffle）,
                一个组的数据在一个分区中，但并不是说一个分区中只有一个组，分组函数会根据计算出来的key进行分组
      2、mapValues():将key-value中的数据取出来，只取values，不取key
      3、groupByKey():相同的key会被分配到一个组中，将数据源中的数据，相同key的数据分在一个组中，形成一个对偶元组，
                    元组中的第一个元素就是key，元组中的第二个元素就是相同key的value的集合
      4、reduceByKey():先对分区内的数据进行key值相同的数据进行聚合，然后再进行shuffle操作，性能比groupByKey好，分区内和分区间的规则要相同
      5、aggregateByKey(2)(_+_,_+_):将数据根据不同的规则进行分区内计算和分区间计算，存在两个柯里化的参数列表：
                            第一个参数列表表示：
                                  需要传递一个参数，表述为初始值，主要用于当碰见第一个key时，和value进行分区内计算
                            第二个参数列表表示：
                                 第一个参数表示分区内的计算操作
                                 第二个参数表示分区间的计算操作
      6、foldByKey(2)(_+_):用于分区内和分区间计算规则相同，参数只有初始值和计算规则的值
      7、combineByKey():通常是把参与运算的第一个数做一些格式上的转换，需要三个参数，并且转换过程需要进行动态识别，所以要加上参数类型
                       第一个参数表示：将相同key的第一个数据进行结构的转换，实现操作
                      第二个参数：分区内的计算规则
                      第三个参数表示：分区间的计算规则
      8、countByKey():统计key-value列表中key的出现次数，返回Map形式
      9、countByValue():统计列表中某个值出现的次数，即有相同的值就自动统计，返回Map形式

       *
   /**
    save相关算子:
          rdd.saveAsTextFile("output"):保存成 Text 文件
          rdd.saveAsObjectFile("output1"):序列化成对象保存到文件
          rdd.map((_,1)).saveAsSequenceFile("output2"):保存成 Sequencefile 文件,要求数据必须是key-value类型
   */
    *
    /**
   RDD的方法和scala方法不一样,
   集合对象的方法都是在同一个节点的内存中完成的
   RDD的方法可以将计算逻辑发送到Excutor端（分布式节点）执行
   RDD和各类共同使用时，需要类实现Serializable序列化
   RDD算子中传递的函数是会包含闭包操作，那么就会进行检测功能
   */
CNGUA01NB07401
   /**
    新的RDD依赖于旧的RDD，依赖关系称为血缘关系
     提高容错性：需要将RDD的依赖关系进行保存，如果错了，则根据依赖关系重新执行
     toDebugString()：可以打印血缘关系，即依赖（其中的+-号表示存在shuffle操作）
     dependencies():打印两个RDD之间的依赖关系
     新的RDD的一个分区数据依赖于旧的RDD的多个依赖
     窄依赖：一对一依赖，旧的RDD的数据只能被一个新的RDD数据进行使用，OneToOneDependency类，执行时分区只有两个，任务也只有两个
     宽依赖:一个父依赖能被多个子依赖使用，中间肯定存在shuffle操作，task数量多，分区被打散
      当RDD中存在shuffle依赖时，阶段会自动增加一个
      阶段的数量=shuffle依赖的数量+1
      ResultStage只有一个，最后需要执行的阶段(一个应用程序会对应多个job)
      任务的数量=当前阶段中最后一个RDD的分区数量
      RDD对象可以重用，但数据不可以重用

      cache()缓存：RDD默认会将数据以缓存的形式存储在内存堆中
      persist(StorageLevel.MEMORY_AND_DISK_2):更改存储级别
      持久化操作需要在执行行动算子之后才能执行
      RDD对象的持久化操作：可以用在数据的重用或者重要数据的持久化
   sc.setCheckpointDir(""):设置保存路径，一般是保存在分布式存储系统HDFS中
   RDD.checkpoint():RDD内存文件需要落盘，需要指定检查点的路径。检查点路径保存的文件，当作业执行完毕后，不会被删除
   cache:将数据临时存储在内存中进行数据重用，cache的底层就是persist
persist:将数据临时存储在磁盘文件中进行数据重用，涉及到磁盘IO，性能较低，但是数据安全，如果作业执行完成，临时保存的数据文件就会丢失。
checkpoint:将数据长久的保存在磁盘文件中进行数据重用，设计到磁盘IO，性能较低，但是数据安全，为了保证数据安全，在一般情况下，会独立执行作业。为了能够提高效率性能，一般会和cache联合使用。
提高效率并保证数据安全：先RDD.cache(),再RDD.checkpoint()

   /**
   RDD文件读取文件：
      1、textFile()读取的是以saveAsTextFile()形式保存的文件
      2、objectFile()读取的是以saveAsObjectFile()形式保存的文件
      3、sequenceFile()读取额是以svaeAsSequenceFile()形式保存的文件
   */

   /**
   自定义累加器：类需要继承AccumulatorV2并设定泛型,并且要重写其中的五个抽象方法
   //自定义数据累加器,
  /**需创建累加器对象，向spark进行注册
   1、定义累加器需要继承AccumulatorV2
   2、定义累加器中的泛型（IN：累加器输入的数据类型，OUT:累加器返回的数据类型）
   3、
    * */
   */
    *
    /**
   广播变量：

   */
    * sql语句查询
    *查看某列数据及某列数据加1：涉及到列运算的时候，每列数据都必须使用$，或者采用引号表达式（单引号+字段名）
    *       df.select('name+1)或df.select($"name>20").show
    *过滤某些数据：df.filter($"age>30").show
    *按照某字段分组，查看每条数据的总数：df.groupBy("age").count.show
    *
    *将现成的RDD转换成数据源：
   *    rdd.toDF("lie1","lie2"):用toDF把RDD转换成数据源的时候需要指定每列的名称
   *    df.rdd:直接把DataFrame数据源转换成RDD数据
   *
   * DataSet数据类型：包含数据类型，属性字段等；
   * DataFrame数据类型：只包含属性字段和属性值
   * RDD：包含属性值
   *其他数据转换成DataSet类型：需要指定数据类型，属性字段，list.toDS
   * DataFrame数据类型转换成DataSet数据类型：df.as(数据结构类型)
   * DataSet转换成DataFrame类型：ds.toDF就可
   *如果rdd构建的时候既有样例类又有属性值，则rdd可以直接转换为DataSet:rdd.toDS
   *
   * 数据转换成DataFrame
   *在使用DataFrame时，如果涉及到转换，需要加入转换规则（引入 import spark.implicits._）
   *在做数据加减时需要引入import xxx.implicits._
   *
   *用户可以通过spark.udf功能添加自定义函数，实现自定义功能
   *      需要用spark.udf.register("Name",(hanshu)=>{值})
   *      调用时需要在要加字段的属性名前面加上相同Name
   *用户可以通过UDAF自定义聚合函数，一般指强类型，强类型聚合函数需要继承Aggregator
   *      ,并实现它的6个方法
   *
   *
   *spark.read.load("lujing"):加载数据的通用方法，需传入数据的路径
   * sparkSql默认读取和保存的文件格式为parquet
   *读取什么格式就在format("geshi")里面加上：spark.read.format("json").load("lujing.json")
   *或者直接spark.read.json("json lujing")
   *通用保存数据：df.write.save()
   *
   * sparksql中访问hive有两种方式：
   * 1、spark sql中包含内置的hive数据库
   * 2、外置hive链接（教程搜一下，注意添加hive依赖，再在配置文件中复制添加hive-site.xml）
   *
   * 实时数据处理：毫秒级别
   * 离线数据处理：小时或者天级别
   * 流式处理：来一条处理一条
   * 批量处理：一批一批处理
   * SparkStreaming：准实时（秒，分），微批次：通过DStream采集器采集数据
   * SparkStreaming两个环境变量：一个是SparkConf()环境配置，另一个是处理周期Second(3)
   * SparkStreaming采集的是长期执行的任务，所以不能直接关闭，main方法执行完毕也会自动关闭
   *ssc.start()
   *ssc.awaitTermination():采用此方法SparkStreaming不会自动关闭
   *用queue采集RDD数据源：
   *        ssc.queueStream(queueOfRDDs)来创建DStream，每一个推送到队列中的RDD，都会作为DStream处理
   *
   *自定义数据源：
   *    需要继承Receiver,并实现onStart、onStop方法用来实现采集时要做的事情
   *    调用时用ssc.receiverStream(new MyRecevice())，其继承了SocketReceiver
   *
   * Kafka数据源：
   *   (kafka仓库读取不到，后面再补充)
   *
   *演示无状态数据操作，只对当前周期内的数据进行处理
   *即像reduceByKey这样的算子
   *采用updateStateByKey()算子可以将当前接受的数据进行状态保留，根据key对数据的状态进行更新，方便下一阶段进行汇总
   *updateStateByKey()：根据key对数据的状态进行更新
   *     第一个参数值：表示相同的key的value数据
   *     第二个参数值：表示缓存区相同key的value数据
   *     使用有状态操作时需要设定检查点路径
   *
   *transform():把一个RDD变成另一个RDD，也可以把底层的RDD获取到进行操作
   *    应用场景：DStream代码不完善，需要代码进行周期性执行
   *    除应用场景外，一般用map进行结构的转换
   *
   *join()：用于在两个流的批次大小一致时，对当前批次的两个流中各自的RDD进行join操作
   *p195
   *
   *
   *
   *
   *
   *
   *
   *
   *

/**
 通信：采用Netty框架，AIO（异步非阻塞式IO），Linux采用Epoll方式模仿AIO操作

    Spark Sql:使用相同的方式访问数据源
        DataFrame是以RDD为基础的分布式数据集，类似于传统二维表格，只是它带有schema元信息（即每一列都带有名称和类型）
        SparkSession是创建DataFrame和执行SQL的入口
        创建spark数据源方式：由数据源创建，从一个存在的RDD进行转换，从Hive Table进行查询

   如果从内寸中读取数据，spark可以知道数据类型是啥，数字默认为Int处理，
   从文件中读取的东西，不能确定类型是啥，只能用bigint接收，bigint只能和Long类型进行转换，不能和Int类型进行转换
   创建一个DataFrame全局表:df.createOrReplaceGlobalTempView("biao"),查询全局表时要在表前面加上global_temp.biao
   创建一个局部临时的DataFrame表:df.createOrReplaceTempView("biao")
   (spark中View只能查询结果集，Table可以增加删除修改)
   查询表中的数据：spark.sql("sql语句").show
   修改数据源数据：df.select($"age"+1).show 或者 df.select('age>20).show
   按照某个字段进行分组，查看数据条数：df.groupBy("age").count.show

   ／

}
